package opc.read.sycread;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.SneakyThrows;
import opc.read.OpcQie.ItemSourceQieUtil;
import opc.read.OpcQie.SingThread;
import opc.read.config.PathConf;
import opc.read.kafka.KafkaSelfProducer;
import opc.read.utils.*;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.jinterop.dcom.common.JIException;
import org.openscada.opc.lib.common.ConnectionInformation;
import org.openscada.opc.lib.common.NotConnectedException;
import org.openscada.opc.lib.da.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @Description: 输入信号-》opc转换一下-》输出信号， 模拟仙鹤的opcserve信号，同步读取，多线程的重构代码
 * @author: liyuan
 * @data 2021-01-06 10:21
 */
public class ByReadUseJson2Multi {
    private static final Logger logger = LoggerFactory.getLogger(ByReadUseJson2Multi.class);
    private static AccessBase access=null;
    public static void DemoAPI() throws IOException, JIException, InterruptedException {
//        final SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");
        final SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd-HH");
// 读取json信号文件，得到字符串。将字符串转成LinkedHashMap
        LinkedHashMap<String, Object> xinhaoSources = ReadJSONItem.xianHe();
        final ConnectionInformation ci = OpcServerUtil.connection();

        final Server server = new Server(ci, Executors.newSingleThreadScheduledExecutor());
        final AutoReconnectController autoReconnectController = new AutoReconnectController(server);
        // 存放模拟或者说实际的信号源  输入和输出json的信号源的key是一样的
        final Map<String, Map<String, Item>> itemSource = new LinkedHashMap<String, Map<String, Item>>();
        // 存放结果
        final MyCopyOnWriteMap<String, Object> result = new MyCopyOnWriteMap<String, Object>(xinhaoSources);

        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
        ScheduledExecutorService writerExecutor = Executors.newScheduledThreadPool(1);
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
//        AccessBase access = null;
        try {
            // connect to server
            autoReconnectController.connect(); //连接opcserver
            access = new SyncAccess(server, 1000); // 获取信号的对象，这里使用同步方式
//            final AccessBase access = new Async20Access(server, 1000, false); // 异步方式  // Failed to change state (true).   changed()方法不被调用，导致result一直不更新
            //server中的map和输入的map转换
            OpcServerUtil.fileIn2OpcIn(itemSource, xinhaoSources, server);
        } catch (Exception e) {
            logger.error("not connection");
            logger.error("出现未连接异常，系统自启\n", e);
            //设置重启
            try {
                //5分钟之后重新连接
                Thread.sleep(1000 * 60 * 1);
                String cmd = PathConf.bathPath;
                Runtime.getRuntime().exec(cmd);
                // Runtime.getRuntime().exec需要延时一下
                Thread.sleep(1000);
                System.exit(0);
            } catch (Exception thE) {
                thE.printStackTrace();
            }
            e.printStackTrace();
        }
        final Map<String, Long> times = new LinkedHashMap<String, Long>();
        final AtomicBoolean timeFlag = new AtomicBoolean(true);
        //首先遍历读一遍
        result.put("date", System.currentTimeMillis());
        /***输入到opcserver的信号值 **/
        for (final String groupName : itemSource.keySet()) {
            Map<String, Item> itemMap = itemSource.get(groupName);
            //组内部的读取信号
            for (Map.Entry<String, Item> map : itemMap.entrySet()) {
                if (map.getValue() instanceof Item) {
                    Item item = map.getValue();
                    Object object = item.read(false).getValue().getObject();
                    Short quality = item.read(false).getQuality();
                    // 当信号质量值小于192的时候，表示该信号有问题，该信号值赋值为null
                    if (quality < 192) {
                        object = null;
                    }
                    // 去掉xianhe.设备 1., 只留下需要的信号名
                    String s = item.getId().split("\\.")[2];
                    /*** 从opcserver输出的信号值 转换一下 存到result中 **/
                    String itemKey = ItemKey2Convert.in2Out(s);
                    Map<String, Object> itemMap2 = (Map<String, Object>) result.get(groupName);
                    if (itemMap2.containsKey(itemKey)) {
                        itemMap2.put(itemKey, object);
                    } else {
                        logger.error("键值对出错：{},{}", s, itemKey);
                    }
                    //计算采样时间的测试代码时间
                    if (PathConf.isTestEnv) {
                        if (item.getId().equals("xianhe.设备 1.PT28851A")) {
                            times.put("PT28851A" + times.size(), System.currentTimeMillis());
                        } else if (item.getId().equals("xianhe.设备 1.PM28_SET_WIDTH")) {
                            times.put("PM28_SET_WIDTH" + times.size(), System.currentTimeMillis());
                        }
                    }

                }
            }
        }

        final Producer producer = KafkaSelfProducer.getProducer();
        final Gson gson = new GsonBuilder().enableComplexMapKeySerialization().create();
        //新开线程开始读取
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                result.put("date", System.currentTimeMillis());
                /***格式化之后没有按照存入的顺序，所以使用gson***/
//                    String jsonString = JSON.toJSONString(result);
                String jsonString = gson.toJson(result);
                logger.info("发送数据到kafka...");
                try {
                    Future future = producer.send(new ProducerRecord<String, String>("zhenxing0", jsonString));//Future future =
                    /** future会出现异常的 Future<RecordMetadata> future =  **/
                    RecordMetadata zhenxing0 = (RecordMetadata) future.get();
                    logger.info("send to topic：" + zhenxing0.topic() + "partition：" + zhenxing0.partition() + "offset：" + zhenxing0.offset());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, 2000, 500, TimeUnit.MILLISECONDS);//单位微妙

        writerExecutor.scheduleAtFixedRate(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                if (PathConf.isTestEnv) {
                    /***格式化之后没有按照存入的顺序，所以使用gson***/
//                           String jsonString = JSON.toJSONString(result);
                    String jsonString = gson.toJson(result);
                    logger.info("--json---\n{}", jsonString);
                    logger.info("将结果保存到文本文件中...");
                    WriterFile.writeTxtFile(jsonString, new File(PathConf.filePath + formatter.format(new Date()) + ".txt"), "utf-8");
                    if (times.size() >= 2 && timeFlag.get() == true) { /***用于计算采集一遍需要的时间 **/
                        WriterFile.writeTxtFile(gson.toJson(times), new File(PathConf.timeJar + formatter.format(new Date()) + ".txt"), "utf-8");
                        timeFlag.set(false);
                    }
                }
            }
        }, 2000, 500, TimeUnit.MILLISECONDS);//单位微妙
        /** 思路将源数据分割，提交到线程中去执行 **/
        // 因为有另外一个线程活着，所以这个线程会一直工作，不会退出。当没有起另外一个线程时，不加while(true)，程序就会退出
        List<Map<String, Map<String, Item>>> itemSourceZu = ItemSourceQieUtil.getItemSourceZu(itemSource);

        for (int i = 0; i < 5; i++) {
            int finalI = i;
            fixedThreadPool.submit(() -> {
                ThreadCai(itemSourceZu.get(finalI), result, access);
            });
        }
        /***
         Thread t1 = new Thread(() -> {
         ThreadCai(itemSourceZu.get(0), result, access);
         }, "t1");
         Thread t2 = new Thread(() -> {
         ThreadCai(itemSourceZu.get(1), result, access);
         }, "t2");
         Thread t3 = new Thread(() -> {
         ThreadCai(itemSourceZu.get(2), result, access);
         }, "t3");
         Thread t4 = new Thread(() -> {
         ThreadCai(itemSourceZu.get(3), result, access);
         }, "t4");
         Thread t5 = new Thread(() -> {
         ThreadCai(itemSourceZu.get(4), result, access);
         }, "t5");
         t1.start();            t2.start();            t3.start();            t4.start();            t5.start();
         ***/
        System.out.println("---main execute----");
        Thread.sleep(1000);
//            access.unbind();

    }

    /**
     * 这个方法可以作为公共方法，因为是引用
     **/
    private static void ThreadCai(Map<String, Map<String, Item>> itemSource, MyCopyOnWriteMap<String, Object> result, AccessBase access) {
        for (final String groupName : itemSource.keySet()) {
            Map<String, Item> itemMap = itemSource.get(groupName);
            //组内部的读取信号
            for (Map.Entry<String, Item> map : itemMap.entrySet()) {
                if (map.getValue() instanceof Item) {
                    //读取单个数据的代码
                    try {
                        access.addItem(map.getValue().getId(), new DataCallback() {
                            /*** DataCallback()是信号值变化的事件回调函数  从结果来看，没有改变的信号也在输出  **/
                            @SneakyThrows
                            @Override
                            public void changed(Item item, ItemState state) {
                                String tmpKey = item.getId().split("\\.")[2];
                                String itemKey = ItemKey2Convert.in2Out(tmpKey); /** 从opcserver出来的值转换一下 **/
                                Map<String, Object> itemMap = (Map<String, Object>) result.get(groupName);
                                if (itemMap.containsKey(itemKey)) {
                                    itemMap.put(itemKey, state.getValue().getObject());  // 去掉xianhe.设备 1., 只留下需要的信号名
//                                        System.out.println("itemKey: " + itemKey + ", state.getValue().getObject(): " + state.getValue().getObject());
                                } else {
                                    StringBuilder builder = new StringBuilder();
                                    builder.append("键值对出错：");
                                    builder.append(tmpKey);
                                    builder.append(",");
                                    builder.append(itemKey);
                                    logger.info(Thread.currentThread().getName() + builder.toString());
                                }
                            }
                        });
                    } catch (JIException | AddFailedException e) {
                        e.printStackTrace();
                    }
                }
                access.bind();
            }
        }
    }
}